package com.casic.kafka.service;

import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;

import com.alibaba.druid.support.json.JSONUtils;
import com.casic.dao.IotDeviceDao;
import com.casic.exception.AopException;
import com.casic.kafka.listener.KafkaSendResultHandler;

import lombok.extern.slf4j.Slf4j;
import net.sf.json.JSONObject;
@Service
@Slf4j
public class KafkaService {
	/**
	 * 线程池
	 */
	 ExecutorService pool = Executors.newFixedThreadPool(1000);
	 
	 /**
	  * kafka执行回调
	  */
	 @Autowired
	 private KafkaSendResultHandler producerListener;
	 
	 /**
	  * 设备的dao
	  */
	 @Autowired
     private IotDeviceDao iot_DeviceDao; 
	 
	 @Autowired
     private KafkaTemplate<String, String> kafkaTemplate;
	
	/**
	 * 写入kafka消息
	 * @param topic
	 * @param message
	 * @return
	 */
	public String writeToKafka(String message) {
		//启动新的线程处理存储kafka
    	pool.submit(new Runnable() {
			@Override
			public void run() {
		
				log.info("Consumer传过来的数据message:"+message.toString());
				
				JSONObject mes = null;
				String equipment = null;
				Long orgId = null;
				//设备ID判断
				try {
					mes = JSONObject.fromObject(message);		
					equipment = (String) mes.get("equipment");
					log.info("获取到的设备id为：" + equipment);
				} catch (Exception e) {
					log.info("获取设备ID异常");
					throw new AopException(e.getLocalizedMessage());
				}
				//设备对应企业判断
				try {
					orgId = iot_DeviceDao.selectOrgIdByEquipment(equipment);
					log.info("获取到的企业id为：" + orgId);
				} catch (Exception e) {
					log.info("设备ID"+equipment+"获取到的企业id异常");
					throw new AopException(e.getLocalizedMessage());
				}
				
				String iot = (String) mes.get("iot");
				log.info("获取iot:"+iot);
				
				HashMap<String, Object> mqttArg = new HashMap<String, Object>();
				mqttArg.put("createTime", String.valueOf(new Date().getTime()));
				mqttArg.put("iot", iot);
				mqttArg.put("equipment", equipment);
				mqttArg.put("orgId", String.valueOf(orgId));
				log.info("设备信息: \n iot:"+iot+"\t equipment:"+equipment+"\t orgId:"+orgId);
				
				
				@SuppressWarnings("unchecked")//对于list类型的可不检查
				List<Map<String, Object>> list = (List<Map<String, Object>>) mes.get("data");
				//类型标示字段
				boolean flag = false;
				//判断是否是状态数据
				for (Map<String, Object> map : list){
					//通过state关键字来判断是状态数据还是实时数据
					if(map.containsKey("state")){
						flag = true;
					}
				}
				/**
				 * 组装kafka所需数据
				 */
				for (Map<String, Object> map : list) { //循环数据的集合
					mqttArg.put("t", map.get("t"));//获得T
				    for(String key :map.keySet()){ //处理其他属性
				    	if(!key.equals("t")) {
				    		mqttArg.put("k",key);
				            mqttArg.put("v",map.get(key));
				            String jsonResult = JSONUtils.toJSONString(mqttArg);
				            if(flag){
								// 存kafka
								//状态数据
								send("IOT_DS_DATA", jsonResult);
				            }else{
								// 存kafka
								//实时数据
					            send("IOT_RT_DATA", jsonResult);
				            }
				    	}
				    }
				    
				}
			}
		});
		return null;
	}
	
	
	/**
	 * 推送消息
	 * @param topic
	 * @param message
	 */
	public void send(String topic, String message) {
		log.info("topic:"+topic+"\n message:"+message);
		try {
			/**
			 * 设置回调监听
			 */
			kafkaTemplate.setProducerListener(producerListener);
			//实时数据
			kafkaTemplate.send(topic, message);
		} catch (Exception e) {
			throw new AopException("\n 推送消息异常  \n "+e.getLocalizedMessage());
		}
		
	}

}

